package com.zst.amqp.config;

import com.zst.amqp.MessageDispatcher;
import com.zst.amqp.MessageListener;
import com.zst.amqp.MessageSender;
import com.zst.amqp.PublisherConfirmCallbackContext;
import com.zst.amqp.entity.ConsumerInfo;
import com.zst.amqp.entity.ConsumerInfoContext;
import com.zst.amqp.utils.ConsumerScanUtil;
import com.zst.amqp.utils.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Lazy;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Spring Amqp与RabbitMQ组件配置类
 * 用于初始化相关组件
 *
 * @author zst
 */
public abstract class BaseAmqpConfiguration {
    private static final Logger logger = LoggerFactory.getLogger(BaseAmqpConfiguration.class);
    /**
     * 交换器名称
     */
    @Value("${app.message.exchangeName:}")
    private String exchangeName;

    /**
     * 队列名称
     */
    @Value("${spring.application.name:}")
    private String queueName;

    /**
     * 扫描Consumer注解的包路径
     */
    @Value("${app.message.scanPackage:}")
    private String basePackage;

    /**
     * 消息消费失败时的重试次数，超过次数之后就不会重新入队列了
     */
    @Value("${app.message.maxRetry:3}")
    private int maxRetry;

    /**
     * 消息重试的时间间隔
     */
    @Value("${app.message.retryInterval:5000}")
    private int retryInterval;

    @Autowired
    private ApplicationContext applicationContext;
    @Autowired
    private AmqpAdmin amqpAdmin;

    /**
     * 创建ConfirmCallbackContext Bean
     */
    @Bean
    public PublisherConfirmCallbackContext confirmCallbackContext() {
        return new PublisherConfirmCallbackContext();
    }

    /**
     * 创建MessageSender Bean
     */
    @Bean
    public MessageSender messageSender(PublisherConfirmCallbackContext confirmCallbackContext,
                                       ConnectionFactory connectionFactory) {
        return new MessageSender(connectionFactory, confirmCallbackContext, this.exchangeName);
    }

    /**
     * 扫描指定路径下的@Consumer注解，创建ConsumerInfoContext实例并注册所有扫描到的注解
     * 如果指定路径下没有扫描到有方法使用了@Consumer注解，则返回null
     * @return
     */
    @Bean
    public ConsumerInfoContext consumerInfoContext() {
        if (StringUtils.isEmpty(basePackage)) {
            return null;
        }

        List<ConsumerInfo> consumerInfos = ConsumerScanUtil.scan(applicationContext, basePackage);
        if (!consumerInfos.isEmpty()) {
            ConsumerInfoContext consumerInfoContext = new ConsumerInfoContext();
            // 注册ConsumerInfo到上下文中
            consumerInfos.forEach(consumerInfo -> {
                try {
                    consumerInfoContext.register(consumerInfo);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });

            return consumerInfoContext;
        }

        return null;
    }

    /**
     * 初始化监听消息的上下文
     * 由于MessageListenerContainer并不需要被别人使用，所以不注册为bean，在此类中保留一个引用就行
     *
     * 这里本来使用@PostConstruct来触发MessageListenerContainer的创建并在此类中保留一个引用的，
     * 但是这样的话导致bean的循环引用，所以只能作为一个bean返回，让spring去处理依赖的先后顺序
     */
    @Bean
    @ConditionalOnBean({ConsumerInfoContext.class, MessageSender.class})
    public MessageListenerContainer messageListenerContainer(ConsumerInfoContext consumerInfoContext, MessageSender messageSender) {
        // 如果不存在ConsumerInfoContext实例，说明程序中没有消费者，不做处理
        if (consumerInfoContext == null) {
            return null;
        }

        // 创建队列和交换器，以及死信重试队列
        Queue queue = declareQueue();
        TopicExchange exchange = declareTopicExchange();
        createDeadLetterExchangeAndQueue();

        // 注册queue和exchange的绑定关系
        for (String msgType : consumerInfoContext.getConsumerMsgTypeSet()) {
            amqpAdmin.declareBinding(BindingBuilder.bind(queue).to(exchange).with(msgType));
        }

        MessageDispatcher messageDispatcher = createMessageDispatcher(consumerInfoContext);
        MessageListener messageListener = createMessageListener(messageSender, messageDispatcher, this.maxRetry, getDeadLetterExchangeName());
        return createMessageListenerContainer(messageListener,
                this.queueName, this.applicationContext.getBean(ConnectionFactory.class));
    }

    //======================================
    // RabbitMQ消费者上下文初始化相关方法
    //======================================
    /**
     * 声明并创建一个Rabbitmq Exchange
     * @return
     */
    private TopicExchange declareTopicExchange() {
        TopicExchange exchange = new TopicExchange(this.exchangeName, true, false);
        this.amqpAdmin.declareExchange(exchange);
        return exchange;
    }

    /**
     * 声明并创建一个Rabbitmq Queue
     * @return
     */
    private Queue declareQueue() {
        // 指定死信队列
        Map<String, Object> properties = new HashMap<>();
        properties.put("x-dead-letter-exchange", getDeadLetterExchangeName());
        Queue queue = new Queue(this.queueName, true, false, false, properties);
        this.amqpAdmin.declareQueue(queue);
        return queue;
    }

    /**
     * 创建当前队列的死信交换机与队列
     */
    private void createDeadLetterExchangeAndQueue() {
        FanoutExchange deadExchange = new FanoutExchange(getDeadLetterExchangeName(), true, false);
        this.amqpAdmin.declareExchange(deadExchange);

        Map<String, Object> properties = new HashMap<>();
        properties.put("x-dead-letter-exchange", this.exchangeName);
        properties.put("x-message-ttl", this.retryInterval);

        Queue deadQueue = new Queue(getDeadLetterQueueName(), true, false, false, properties);
        this.amqpAdmin.declareQueue(deadQueue);
        this.amqpAdmin.declareBinding(BindingBuilder.bind(deadQueue).to(deadExchange));
        // 为了让死信队列的消息在过期之后重新回到原来的队列，这里添加一个binding
//        this.amqpAdmin.declareBinding(BindingBuilder.bind(queue).to(exchange).with(this.queueName));
    }

    /**
     * 消息监听者实例
     * @param messageSender 消息发送者
     * @param messageDispatcher 消息分发者
     * @param maxRetry 最大重试次数
     * @param deadExchangeName 死信队列名称
     * @return
     */
    private MessageListener createMessageListener(MessageSender messageSender, MessageDispatcher messageDispatcher, int maxRetry, String deadExchangeName) {
        return new MessageListener(messageSender, messageDispatcher, maxRetry, deadExchangeName);
    }

    /**
     * 创建MessageListenerContainer
     * @return
     */
    private MessageListenerContainer createMessageListenerContainer(MessageListener messageListener, String queueName, ConnectionFactory connectionFactory) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames(queueName);
        container.setMessageListener(messageListener);
        //这里由于手动创建了MessageListenerContainer，所以没有读取配置文件中的设置值，这里手动指定为手动确认模式
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setAutoDeclare(true);
        // 自己创建的MessageListenerContainer需要手动启动
        container.start();
        return container;
    }

    /**
     * 创建消息分发者实例
     * @return
     */
    private MessageDispatcher createMessageDispatcher(ConsumerInfoContext consumerInfoContext) {
        return new MessageDispatcher(consumerInfoContext, this.applicationContext);
    }

    //=====================================
    // private
    //=====================================
    /**
     * 获取本服务所使用的死信交换器的名称
     * @return
     */
    private String getDeadLetterExchangeName() {
        return this.queueName + "_dlx";
    }

    /**
     * 获取本服务所使用的死信队列的名称
     * @return
     */
    private String getDeadLetterQueueName() {
        return this.queueName + "_dlq";
    }
}
